package com.by.mqsender.support;

import cn.hutool.core.util.IdUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.by.mqsender.config.ByMqConfig;
import com.by.mqsender.consts.ByMqConst;
import com.by.toolkit.consts.ByContextConst;
import com.by.toolkit.platform.ByContext;
import com.by.toolkit.utils.ByIfNotNullUtil;
import com.by.toolkit.utils.ByIfTrueUtil;
import com.by.toolkit.utils.SplitterUtil;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
import org.apache.rocketmq.client.java.message.MessageBuilderImpl;

import javax.annotation.PreDestroy;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;

import static com.by.toolkit.utils.ExceptionUtil.getStackTrace;
import static com.by.toolkit.utils.ExceptionUtil.nullError;

/**
 * 消息队列发送者
 *
 * @author YanBingHao
 * @since 2025/9/29
 */
@Slf4j
public class ByMqSupport {

    private Map<String, Producer> producerMap;

    private ByMqConfig byMqConfig;

    public ByMqSupport(Map<String, Producer> topicProducerMap, ByMqConfig byMqConfig) {
        this.producerMap = topicProducerMap;
        this.byMqConfig = byMqConfig;
    }

    public static final Map<String, String> COPY_PROPERTY = ImmutableMap.of(
            ByContextConst.CORP_CODE, ByMqConst.CORP_CODE,
            ByContextConst.PROJECT_ID, ByMqConst.PROJECT_ID
    );

    public static final Map<String, String> MQ_COPY_PROPERTY = COPY_PROPERTY.entrySet()
            .stream()
            .collect(Collectors.toMap(Map.Entry::getValue, Map.Entry::getKey));

    private Map<String, String> getCopyProperty() {
        return COPY_PROPERTY.entrySet()
                .stream()
                .filter(i -> StringUtils.isNotBlank(ByContext.get(i.getKey())))
                .collect(Collectors.toMap(Map.Entry::getValue, i -> ByContext.get(i.getKey())));
    }


    /**
     * 异步发送普通消息
     *
     * @param body
     * @param tag
     */
    public void sendAsync(Map<String, Object> body, @NonNull String tag) {
        String topic = StringUtils.defaultIfBlank(byMqConfig.tag2Topic(tag), ByMqConst.DEFAULT_TOPIC);
        this.sendAsync(body, topic, tag, null, null, null);
    }

    public void sendAsync(Object body, @NonNull String tag) {
        String topic = StringUtils.defaultIfBlank(byMqConfig.tag2Topic(tag), ByMqConst.DEFAULT_TOPIC);
        this.sendAsync(JSONObject.from(body), topic, tag, null, null, null);
    }

    /**
     * 异步发送普通消息
     *
     * @param body
     * @param tag
     * @param callback
     */
    public void sendAsyncWithCallback(Map<String, Object> body, @NonNull String tag, MqSendCallback callback) {
        String topic = StringUtils.defaultIfBlank(byMqConfig.tag2Topic(tag), ByMqConst.DEFAULT_TOPIC);
        this.sendAsync(body, topic, tag, null, null, callback);
    }

    /**
     * 异步发送顺序消息
     * <p>
     * https://help.aliyun.com/zh/apsaramq-for-rocketmq/cloud-message-queue-rocketmq-5-x-series/developer-reference/ordered-messages-1?spm=a2c4g.11186623.0.0.71dd7a62ZamHn6#p-dbv-ptn-5id
     *
     * @param body
     * @param tag
     */
    public void sendFifoAsync(Map<String, Object> body, @NonNull String tag, String messageGroup) {
        String topic = StringUtils.defaultIfBlank(byMqConfig.tag2Topic(tag), ByMqConst.DEFAULT_PRODUCER);
        this.sendAsync(body, topic, tag, messageGroup, null, null);
    }

    /**
     * 异步发送顺序消息
     * <p>
     * https://help.aliyun.com/zh/apsaramq-for-rocketmq/cloud-message-queue-rocketmq-5-x-series/developer-reference/ordered-messages-1?spm=a2c4g.11186623.0.0.71dd7a62ZamHn6#p-dbv-ptn-5id
     *
     * @param body
     * @param tag
     */
    public void sendFifoAsyncWithCallback(Map<String, Object> body, @NonNull String tag, String messageGroup, @NonNull MqSendCallback callback) {
        String topic = StringUtils.defaultIfBlank(byMqConfig.tag2Topic(tag), ByMqConst.DEFAULT_PRODUCER);
        this.sendAsync(body, topic, tag, messageGroup, null, callback);
    }

    /**
     * 异步发送延时消息
     *
     * @param body
     * @param tag
     * @param delayMillis
     */
    public void sendDelayAsync(Map<String, Object> body, @NonNull String tag, @NonNull Long delayMillis) {
        String topic = StringUtils.defaultIfBlank(byMqConfig.tag2Topic(tag), ByMqConst.DEFAULT_DELAY_TOPIC);
        this.sendAsync(body, topic, tag, null, System.currentTimeMillis() + delayMillis, null);
    }

    public void sendDelayAsync(Object body, @NonNull String tag, @NonNull Long delayMillis) {
        String topic = StringUtils.defaultIfBlank(byMqConfig.tag2Topic(tag), ByMqConst.DEFAULT_DELAY_TOPIC);
        this.sendAsync(JSONObject.from(body), topic, tag, null, System.currentTimeMillis() + delayMillis, null);
    }

    /**
     * 异步发送延时消息
     *
     * @param body
     * @param tag
     * @param delayMillis
     */
    public void sendDelayAsyncWithCallback(Map<String, Object> body, @NonNull String tag, @NonNull Long delayMillis, @NonNull MqSendCallback callback) {
        String topic = StringUtils.defaultIfBlank(byMqConfig.tag2Topic(tag), ByMqConst.DEFAULT_DELAY_TOPIC);
        this.sendAsync(body, topic, tag, null, System.currentTimeMillis() + delayMillis, callback);
    }

    /**
     * 异步发送指定时间点的消息
     *
     * @param body
     * @param tag
     * @param timestamp
     */
    public void sendTimepointAsync(Map<String, Object> body, @NonNull String tag, Long timestamp) {
        String topic = StringUtils.defaultIfBlank(byMqConfig.tag2Topic(tag), ByMqConst.DEFAULT_DELAY_TOPIC);
        this.sendAsync(body, topic, tag, null, timestamp, null);
    }

    /**
     * 异步发送指定时间点的消息
     *
     * @param body
     * @param tag
     * @param timestamp
     */
    public void sendTimepointAsyncWithCallback(Map<String, Object> body, @NonNull String tag, @NonNull Long timestamp, @NonNull MqSendCallback callback) {
        String topic = StringUtils.defaultIfBlank(byMqConfig.tag2Topic(tag), ByMqConst.DEFAULT_DELAY_TOPIC);
        this.sendAsync(body, topic, tag, null, timestamp, callback);
    }

    private void sendAsync(Map<String, Object> param, String topic, String tag, String messageGroup, Long timestamp, MqSendCallback callback) {
        nullError(tag, "tag不能为空！");
        nullError(param, "body不能为空！");

        Producer producer = ByIfTrueUtil.getIfTrue(Objects.nonNull(producerMap.get(topic))
                , () -> producerMap.get(topic)
                , producerMap.get(Objects.nonNull(timestamp) ? ByMqConst.DEFAULT_DELAY_PRODUCER : ByMqConst.DEFAULT_PRODUCER));

        Map<String, Object> body = Maps.newHashMap(param);

        Map<String, String> copyProperty = this.getCopyProperty();
        log.info("[ByMqSupport][sendAsync] copyProperty:[{}]", copyProperty);
        if (Objects.nonNull(copyProperty) && !copyProperty.isEmpty()) {
            body.putAll(copyProperty);
        }
        String keys = SplitterUtil.UNDERSCORE_JOINER.join(tag, IdUtil.fastSimpleUUID());

        Message message = ByIfNotNullUtil.getIfNotNullOrElse(timestamp,
                () -> new MessageBuilderImpl()
                        .setTopic(topic)
                        .setTag(tag)
                        .setKeys(keys)
                        .setBody(JSON.toJSONString(body).getBytes(StandardCharsets.UTF_8))
                        .setDeliveryTimestamp(timestamp)
                        .build(),
                () -> ByIfTrueUtil.getIfTrueOrElse(StringUtils.isNotBlank(messageGroup),
                        () -> new MessageBuilderImpl()
                                .setTopic(topic)
                                .setTag(tag)
                                .setKeys(keys)
                                .setBody(JSON.toJSONString(body).getBytes(StandardCharsets.UTF_8))
                                .setMessageGroup(messageGroup)
                                .build(),
                        () -> new MessageBuilderImpl()
                                .setTopic(topic)
                                .setTag(tag)
                                .setKeys(keys)
                                .setBody(JSON.toJSONString(body).getBytes(StandardCharsets.UTF_8))
                                .build())
        );

        ThreadPoolExecutor threadPoolExecutor = byMqConfig.executor();
        ByIfTrueUtil.doIfTrueOrElse(Objects.isNull(threadPoolExecutor)
                , () -> {
                    try {
                        SendReceipt sendReceipt = producer.send(message);
                        if (Objects.nonNull(callback)) {
                            callback.messageId(sendReceipt.getMessageId().toString());
                        }
                        log.info("[ByMqSupport][sendAsync] sendAsyncSuccess topic:[{}] tag:[{}] messageId:[{}]", topic, tag, sendReceipt.getMessageId());
                    } catch (Exception throwable) {
                        log.error("[ByMqSupport][sendAsync] sendAsyncError topic:[{}] tag:[{}] [HasException]:[{}]", topic, tag, getStackTrace(throwable));
                    }
                }
                , () -> {
                    threadPoolExecutor.execute(() -> {
                        try {
                            SendReceipt sendReceipt = producer.send(message);
                            if (Objects.nonNull(callback)) {
                                callback.messageId(sendReceipt.getMessageId().toString());
                            }
                            log.info("[ByMqSupport][sendAsyncExecutor] sendAsyncSuccess topic:[{}] tag:[{}] messageId:[{}]", topic, tag, sendReceipt.getMessageId());
                        } catch (Exception throwable) {
                            log.error("[ByMqSupport][sendAsyncExecutor] sendAsyncError topic:[{}] tag:[{}] [HasException]:[{}]", topic, tag, getStackTrace(throwable));
                        }
                    });
                });
    }


    @PreDestroy
    public void close() {
        log.info("[ByMqSupport][close] mqProducerClosing");
        producerMap.values().forEach(p -> {
            try {
                p.close();
            } catch (Exception e) {
                log.warn("[ByMqSupport][close] closeProducerException:[{}]", getStackTrace(e));
            }
        });
        log.info("[ByMqSupport][close] mqProducerClosed");
    }


    public interface MqSendCallback {

        void messageId(String messageId);

    }


}
